package com.tyc.kafka.controller;

import com.tyc.kafka.common.LogData;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

/**
 * 类描述
 *
 * @author tyc
 * @version 1.0
 * @date 2022-11-29 11:49:43
 */
@RestController
@Slf4j
public class TestController {
    @Autowired
    private KafkaConsumer kafkaConsumer;

    @GetMapping("/start")
    public String consume(){
        Boolean run  = true;
//        Integer max = 17;
//        Integer min = 13;
        Integer max = 29900978;
        Integer min = 27967299;
        // 假如要消费 test 主题 0 分区 offSet 1000 到 1500 的数据
        //kafka的分区逻辑是在poll方法里执行的,所以执行seek方法之前先执行一次poll方法
        Set<TopicPartition> assignment = new HashSet<>();
        while (assignment.size() == 0) {
            //如果没有分配到分区,就一直循环下去
            kafkaConsumer.poll(100L);
            assignment = kafkaConsumer.assignment();
        }
        for (TopicPartition tp : assignment) {
            // 指定消息 game-server-log 与 logs
            if (tp.topic().equals("game-server-log") && tp.partition() == 0) {
                kafkaConsumer.seek(tp, min);
                break;
            }
        }

        while (run){
            ConsumerRecords<String, String> records = kafkaConsumer.poll(1000);
            Iterator<ConsumerRecord<String, String>> iterator = records.iterator();
            while (iterator.hasNext()){
                ConsumerRecord<String, String> next = iterator.next();
                log.info("消费数据====>主题：{},offset:{},num:{}",next.topic(),next.offset(),records.count());
                if(next.offset() > max){
                    run = false;
                }
                break;
            }
            if(!records.isEmpty()){
                for (ConsumerRecord<String, String> record : records) {
                    if(record.offset() <= max){
                        // 执行消费逻辑
                        log.info(record.toString());
                        LogData.record(record.value());
                    }else {
                        run = false;
                    }
                }
            }

        }
        return "SUCCESS";
    }
}
